package wsconn

import (
	"errors"
	"fmt"
	"kubeInstall/pkg/logger"
	"net/http"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

type handleMessageFunc func([]byte)

// Session wrapper around websocket connections.
type Session struct {
	Request *http.Request
	conn    *websocket.Conn
	output  chan *wsMsg
	open    bool
	rwmutex *sync.RWMutex
	Config  *Config

	inComeMessageHandler handleMessageFunc
	outputMessageHandler handleMessageFunc
}

func (s *Session) HandleInComeMessage(fn func([]byte)) {
	s.inComeMessageHandler = fn
}

func (s *Session) HandleOutputMessage(fn func([]byte)) {
	s.outputMessageHandler = fn
}

func NewSession(w http.ResponseWriter, r *http.Request) *Session {
	upgrader := &websocket.Upgrader{
		ReadBufferSize:  1024,
		WriteBufferSize: 1024,
		CheckOrigin:     func(r *http.Request) bool { return true },
	}
	conn, err := upgrader.Upgrade(w, r, w.Header())

	if err != nil {
		return nil
	}

	session := &Session{
		Request: r,
		conn:    conn,
		output:  make(chan *wsMsg, 1024),
		open:    true,
		rwmutex: &sync.RWMutex{},
		Config:  newConfig(),
	}
	return session

}

func (s *Session) Run() {
	go s.writePump()
	go s.readPump()
}
func (s *Session) ping() {
	s.writeRaw(&wsMsg{t: websocket.PingMessage, msg: []byte{}})
}

func (s *Session) writePump() {
	ticker := time.NewTicker(s.Config.PingPeriod)
	defer ticker.Stop()

loop:
	for {
		select {
		case msg, ok := <-s.output:
			if !ok {
				break loop
			}

			err := s.writeRaw(msg)

			if err != nil {
				s.errorHandler(err)
				break loop
			}

			if msg.t == websocket.CloseMessage {
				break loop
			}

			if msg.t == websocket.TextMessage {
				s.outputMessageHandler(msg.msg)
			}

		case <-ticker.C:
			s.ping()
		}
	}
}

func (s *Session) readPump() {
	s.conn.SetReadLimit(s.Config.MaxMessageSize)
	s.conn.SetReadDeadline(time.Now().Add(s.Config.PongWait))

	for {
		t, message, err := s.conn.ReadMessage()
		fmt.Println(string(message))
		if err != nil {
			s.errorHandler(err)
			break
		}

		if t == websocket.TextMessage {
			s.inComeMessageHandler(message)
		}

	}
}

func (s *Session) writeMessage(message *wsMsg) {
	if s.closed() {
		s.errorHandler(errors.New("tried to write to closed a session"))
		return
	}

	select {
	case s.output <- message:
	default:
		s.errorHandler(errors.New("session message buffer is full"))
	}
}

func (s *Session) writeRaw(message *wsMsg) error {
	if s.closed() {
		return errors.New("tried to write to a closed session")
	}

	s.conn.SetWriteDeadline(time.Now().Add(s.Config.WriteWait))

	err := s.conn.WriteMessage(message.t, message.msg)

	if err != nil {
		return err
	}

	return nil
}

// Write writes message to session.
func (s *Session) Write(msg []byte) error {
	if s.closed() {
		return errors.New("session is closed")
	}

	s.writeMessage(&wsMsg{t: websocket.TextMessage, msg: msg})

	return nil
}

// Close closes session.
func (s *Session) SendCloseConnSignal() error {
	if s.closed() {
		return errors.New("session is already closed")
	}

	s.writeMessage(&wsMsg{t: websocket.CloseMessage, msg: []byte{}})

	return nil
}

func (s *Session) SendCloseConnSignalWithMsg(msg []byte) error {
	if s.closed() {
		return errors.New("session is already closed")
	}

	s.writeMessage(&wsMsg{t: websocket.CloseMessage, msg: msg})

	return nil
}

func (s *Session) IsSessionClosed() bool {
	return s.closed()
}

func (s *Session) errorHandler(err error) {

	logger.Error("ws conn error :%v", err)
}

func (s *Session) closed() bool {
	s.rwmutex.RLock()
	defer s.rwmutex.RUnlock()

	return !s.open
}

func (s *Session) CloseSession() {
	if !s.closed() {
		s.rwmutex.Lock()
		s.open = false
		s.conn.Close()
		close(s.output)
		s.rwmutex.Unlock()
	}
}

func FormatCloseMessage(closeCode int, text string) []byte {
	return websocket.FormatCloseMessage(closeCode, text)
}
